package ds_industry_2025.ds.ds_06.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

/*
    3、请根据dwd_ds_hudi层的相关表，计算2020年销售量前10的商品，销售额前10的商品，存入ClickHouse数据库shtd_result的topten表
    中（表结构如下），然后在Linux的ClickHouse命令行中根据排名升序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务
    B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下;
 */
object t3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t3")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val detail_path="hdfs://192.168.40.110:9000/user/hive/warehouse/dwd_ds_hudi.db/fact_order_detail"


    //  这一题的订单金额如果直接使用fact_order_info里面的话，那计算商品销售量就会比较麻烦，所以我们完全使用order_detail

    spark.read.format("hudi").load(detail_path).createOrReplaceTempView("detail")
    spark.read.format("hudi").load(detail_path)
      .where("etl_date=(select max(etl_date) from detail)")
      .where(year(col("create_time")) === 2020)
      .createOrReplaceTempView("detail_info")



    //  todo 首先计算商品销售额前十的商品信息
    val r1 = spark.sql(
      """
        |select distinct
        |sku_id as topquantityid,
        |sku_name as topquantityname,
        |total_count topquantity,
        |row_number() over(order by total_count desc ) as sequence
        |from(
        |select distinct
        |sku_id,sku_name,
        |sum(sku_num) over(partition by sku_id,sku_name) as total_count
        |from detail_info
        |) as r1
        |limit 10
        |""".stripMargin)

    //  todo 然后计算销售额前10的
    val r2 = spark.sql(
      """
        |select
        |sku_id as toppriceid,
        |sku_name toppricename,
        |(total_count * order_price) as topprice,
        |row_number() over(order by (total_count * order_price) desc) as sequence
        |from(
        |select distinct
        |sku_id,sku_name,order_price,
        |sum(sku_num) over(partition by sku_id,sku_name) as total_count
        |from detail_info
        |) as r1
        |limit 10
        |""".stripMargin)

    //  todo 将两个结果合并
    r1.join(r2,"sequence")
      .write.format("jdbc")
      .option("url","jdbc:clickhouse://192.168.40.110:8123/shtd_result")
      .option("user","default")
      .option("password","")
      .option("driver","com.clickhouse.jdbc.ClickHouseDriver")
      .option("dbtable","topten")
      .mode("append")
      .save()

    // todo select * from topten order by sequence limit 5;

    spark.close()

  }

}
